
#include <iostream>

#include <pqxx/pqxx> 
using namespace pqxx;

#include "pqxx_transaction.hpp"
#include "pqxx_common.hpp"
#include "pqxx_session.hpp"

namespace topdbs
{
	pqxx_session::pqxx_session()
		: m_pConnection(nullptr)
		, m_pStmt(nullptr)
		, m_pCursor(nullptr)
	{
	}

	pqxx_session::pqxx_session(pqxx_connection* pConnection)
		: m_pConnection(pConnection)
		, m_pStmt(nullptr)
		, m_pCursor(nullptr)
	{
	}

	pqxx_session::~pqxx_session()
	{
		if(m_pStmt)
		{
			delete m_pStmt;
			m_pStmt = nullptr;
		}
		if(m_pCursor)
		{
			delete m_pCursor;
			m_pCursor = nullptr;
		}
	}

	//! reset the current command all info.
	void pqxx_session::Reset() 
	{
		topdbs_base_session::Reset();
	}

	//! Validates and prepares the current command for execution.
	int  pqxx_session::Prepare() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);
		CHK_SQL_ERROR_RETURN(m_commandText,ERR_DBS_ARGS);

		pqxx::connection* connection_ptr = (pqxx::connection*)m_pConnection->getConnection();
		if(m_commandType == eDBSCommandType::DBS_CmdSQLCursor)
		{
			format_placeholders(m_pConnection->getClientType(),m_commandText);
			return Prepare_cursor(connection_ptr);
		}
		else if(m_commandType == eDBSCommandType::DBS_CmdSQLStmt)
		{
			format_placeholders(m_pConnection->getClientType(),m_commandText);
			return Prepare_stmt(connection_ptr);							
		}
		else 
		{
			m_commandFields.clear();
			m_numParams = cvtPrepareParams(m_commandText,m_commandFields);
		}
		m_isPrepared = true;
		return ERR_DBS_OK;
	}

	//! Executes the current command.
	int  pqxx_session::Execute() 
	{
		int nRet = ERR_DBS_OK;
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);
		CHK_SQL_ERROR_RETURN(m_commandText,ERR_DBS_ARGS);

		dbslog(DBS_DEBUG,"commandType=%d commandText=[%s]",m_commandType,m_commandText.c_str());

		m_sqlrows = 0;
		if( m_commandFields.size() > 0 )
		{
			cvtFormatParams(m_commandText,m_commandFields);
		}		
		if(m_commandType == eDBSCommandType::DBS_CmdUnknown)
		{
			nRet = IsSelectSQL(m_commandText) 
				? ExecuteQuery(m_commandText) 
				: ExecuteUpdate(m_commandText);
		}	
		else if(m_commandType == eDBSCommandType::DBS_CmdSQLStmt )
		{
			if(!m_isPrepared)
			{
				m_sqlerr = "Incomplete pretreatment";
				dbslog(DBS_ERROR,"%s",m_sqlerr.c_str());
				return ERR_DBS_FAIL;
			}
			nRet = ExecuteStmt();		
		}
		else if(m_commandType == eDBSCommandType::DBS_CmdSQLCursor)
		{
			if(!m_isPrepared)
			{
				m_sqlerr = "Incomplete pretreatment";
				dbslog(DBS_ERROR,"%s",m_sqlerr.c_str());
				return ERR_DBS_FAIL;
			}
			nRet = ExecuteCursor();
		}
		m_isExecuted = (nRet == ERR_DBS_OK) ? true : false;
		return nRet;	
	}

	int  pqxx_session::Execute(const char* sSQL,...) 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);	
		m_commandText = sSQL;
		va_list vaList;
		va_start(vaList, sSQL);
		int nRet=ExecuteV(sSQL,vaList);
		va_end(vaList);	
		return nRet;
	}	

	int  pqxx_session::ExecuteV(const string& sSQL,va_list vaList) 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);	
		return topdbs_base_session::ExecuteV(sSQL,vaList) == ERR_DBS_OK ? Execute() : ERR_DBS_FAIL;
	}

	int  pqxx_session::ExecuteFetch(const string& sSQL) 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);	
		return IsSelectSQL(sSQL) ? ExecuteQuery_Row(sSQL) : ExecuteUpdate(sSQL);
	}

	bool pqxx_session::FetchNext() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,false);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),false);

		bool isFoundNext = false;
		if(m_commandType == eDBSCommandType::DBS_CmdUnknown)
		{
			return topdbs_base_session::FetchNext();
		}
		else if(m_commandType == eDBSCommandType::DBS_CmdSQLStmt)
		{
			CHK_SESSION_HANDLE_ERROR_RETURN(m_pStmt,false);
			try
			{
				pqxx::result& R = *(m_pStmt->m_resultset);
				if( m_pStmt->m_dataindex  < R.size() )
				{
					
					auto  rowdata = R.at(m_pStmt->m_dataindex);
					for(auto col = rowdata.begin(); col != rowdata.end(); col++)
					{
						int coltyp;
						std::string colna,colval("");
						colna = col->name();
						if(!col->is_null())
						{
							colval = col->c_str();
						}
						m_row.insert(std::pair<std::string, std::string>(topdbs_string::to_lower(colna), colval));
					}
					(m_pStmt->m_dataindex)++;
					isFoundNext = true;	
				}
			}
			catch(const pqxx::pqxx_exception& ex)
			{
				m_sqlerr = ex.base().what();
				dbslog(DBS_ERROR,"FetchNext() Exception: %s !!",m_sqlerr.c_str());		
			}
		}	
		else if(m_commandType == eDBSCommandType::DBS_CmdSQLCursor)
		{
			isFoundNext =  FetchCursorNext();
		}	
		return isFoundNext;
	}

	bool pqxx_session::FetchCursorNext(uint16_t nFetchSize) 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,false);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),false);	
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pCursor,false);

		bool isFoundNext = false;
		if(m_commandType != eDBSCommandType::DBS_CmdSQLCursor)
		{
			return isFoundNext;
		}

		PqxCursor* pqxCuror = m_pCursor;
		try
		{
			//uint16_t  nFetchSize = 1;
			if( !pqxCuror->m_resultset || 
				( pqxCuror->m_resultset && pqxCuror->m_cur_dataindex >= pqxCuror->m_resultset->size() ))
			{
				if( pqxCuror->m_resultset ) 
				{
					pqxCuror->m_resultset->clear();
					*(pqxCuror->m_resultset) = pqxCuror->m_cursor->retrieve( 
						pqxCuror->m_dataindex, 
						pqxCuror->m_dataindex + nFetchSize);
				}
				else
				{
					pqxCuror->m_resultset = new pqxx::result(
						pqxCuror->m_cursor->retrieve( 
							pqxCuror->m_dataindex, 
							pqxCuror->m_dataindex + nFetchSize ));
				}
				pqxCuror->m_dataindex += nFetchSize;
				pqxCuror->m_cur_dataindex = 0;
			}
			pqxx::result& R = *(pqxCuror->m_resultset);
			size_t records = R.size();
			if(records == 0)
			{
				return isFoundNext;
			}

			auto& row = R.at(pqxCuror->m_cur_dataindex);
			pqxCuror->m_cur_dataindex++;

			m_row.clear();
			for(auto col = row.begin(); col != row.end(); col++)
			{
				int coltyp;
				std::string colna,colval("");
				colna = col->name();
				if(!col->is_null())
				{
					colval = col->c_str();
				}
				m_row.insert(std::pair<std::string, std::string>(topdbs_string::to_lower(colna), colval));	
			}
			m_sqlrows++;
		}
		catch(const pqxx::range_error& ex)
		{
			return isFoundNext;
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"FetchCursorNextRow Exception erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}	
		isFoundNext = true;
		m_sqlrows++;
		return isFoundNext;	
	}

	//! Binding the command parameter specified by its position
	int  pqxx_session::BindParams(const uint16_t paramno,const DBSValue& value) 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);	
		string param_value = value.toString();
		if(m_commandType == eDBSCommandType::DBS_CmdSQLStmt)
		{
			CHK_SESSION_HANDLE_ERROR_RETURN(m_pStmt,ERR_DBS_UNKOWN);
			auto& stmt = *m_pStmt->m_stmt;
			stmt(param_value);
		}
		else
		{
			return ReplaceParams(paramno,value);
		}	
		return ERR_DBS_OK;
	}

	int  pqxx_session::BindParams(const uint16_t paramno,const string& value) 
	{
		return BindParams(paramno,DBSValue().setAsString(value));
	}

	int  pqxx_session::BindParams(const uint16_t paramno,const int     value) 
	{
		string para_value = topdbs_string::to_string<int>(value);
		return BindParams(paramno,para_value);
	}

	int  pqxx_session::BindParams(const uint16_t paramno,const long    value) 
	{
		string para_value = topdbs_string::to_string<long>(value);
		return BindParams(paramno,para_value);
	}

	int  pqxx_session::BindParams(const uint16_t paramno,const float   value) 
	{
		string para_value = topdbs_string::to_string<float>(value);
		return BindParams(paramno,para_value);	
	}

	int  pqxx_session::BindParams(const uint16_t paramno,const double  value) 
	{
		string para_value = topdbs_string::to_string<double>(value);
		return BindParams(paramno,para_value);	
	}

	int pqxx_session::Prepare_cursor(pqxx::connection* connection_ptr)
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);
		if(!IsSelectSQL(m_commandText))
		{
			m_sqlerr = "The cursor operation unsupported unquery statement";
			dbslog(DBS_ERROR,"%s !!",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}
		try
		{
			m_pConnection->begin();

			string cursor_name(TOPDBS_DEFAULT_CURSOR_NAME);	
			char now_time[40] = {0};
			topdbs_gettime(now_time);
			cursor_name.append(now_time);		
			if(!m_pCursor)
			{
				m_pCursor = new PqxCursor(
					*m_pConnection->getTransaction(),
					m_commandText,
					cursor_name,
					true);
			}
			m_pCursor->m_connection_ptr = connection_ptr;
			m_sqlrows=0;	
			m_numParams = cvtGetPreParamCount(m_commandText);
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"Prepare_cursor Exception erortext[%s]",m_sqlerr.c_str());
			m_pConnection->rollback();
			return ERR_DBS_FAIL;
		}	
		m_isPrepared = true;
		return ERR_DBS_OK;
	}

	int pqxx_session::Prepare_stmt(pqxx::connection* connection_ptr)
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);
		try
		{
			m_pConnection->begin();

			string prepare_name("myprepare");
			char now_time[40] = {0};
			topdbs_gettime(now_time);
			prepare_name.append(now_time);

			connection_ptr->prepare(prepare_name,m_commandText);
			if(!m_pStmt)
			{
				m_pStmt = new PqxStatement(
					connection_ptr,
					*m_pConnection->getTransaction(),
					prepare_name);
			}
			m_numParams = cvtGetPreParamCount(m_commandText);
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"Prepare_stmt Exception  erortext[%s]",m_sqlerr.c_str());
			m_pConnection->rollback();
			return ERR_DBS_FAIL;
		}
		m_isPrepared = true;
		return ERR_DBS_OK;	
	}

	int pqxx_session::ExecuteStmt()
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pStmt,ERR_DBS_UNKOWN);
		try
		{
			m_pStmt->m_resultset = new pqxx::result(m_pStmt->m_stmt->exec());
			m_sqlrows = m_pStmt->m_resultset->affected_rows();
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"ExecuteStmt Exception erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}
		if(IsSelectSQL(this->m_commandText))
		{
			m_isResultSet = true;
		}		
		return ERR_DBS_OK;		
	}

	int pqxx_session::ExecuteCursor()
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection->getConnection(),ERR_DBS_UNKOWN);
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pCursor,ERR_DBS_UNKOWN);
		try
		{
			m_pCursor->m_cursor = new pqxx::stateless_cursor<pqxx::cursor_base::read_only, pqxx::cursor_base::owned>(
				m_pCursor->m_trans, 
				//m_pCursor->m_sql, 
				m_commandText, 
				m_pCursor->m_cursor_name, 
				true);
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"ExecuteCursor Exception erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}
		if(IsSelectSQL(this->m_commandText))
		{
			m_isResultSet = true;
		}	
		return ERR_DBS_OK;		
	}

	int pqxx_session::ExecuteUpdate(const std::string& sSQL)
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SQL_ERROR_RETURN(sSQL,ERR_DBS_ARGS);
		if(IsSelectSQL(sSQL)) 
		{
			m_sqlerr = "Invalid SQL statement for Update";		
			return ERR_DBS_FAIL;
		}
		try
		{
			if(  m_pConnection->getTransaction() && m_pConnection->isTransaction() )
			{
				m_sqlrows = m_pConnection->getTransaction()->exec(sSQL).affected_rows();
			}
			else
			{
				pqxx::connection* connection_ptr = (pqxx::connection*)m_pConnection->getConnection();
				auto_work work(*connection_ptr);
				m_sqlrows = work.exec(sSQL).affected_rows();
			}
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();		
			dbslog(DBS_ERROR,"ExecuteUpdate Exception erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}
		return ERR_DBS_OK;
	}

	int pqxx_session::ExecuteQuery_Row(const std::string& sSQL)
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SQL_ERROR_RETURN(sSQL,ERR_DBS_ARGS);
		if(!IsSelectSQL(sSQL))
		{
			m_sqlerr = "Invalid SQL statement for Select";
			dbslog(DBS_ERROR,"ExecuteQuery_Row Exception erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}

		try
		{
			pqxx::connection* connection_ptr = (pqxx::connection*)m_pConnection->getConnection();
			nontransaction N(*connection_ptr);
			result R( N.exec( sSQL ));

			m_sqlrows = 0;
			m_row.clear();
			for(result::const_iterator c = R.begin(); c != R.end(); ++c)
			{
				for(auto col = c.begin(); col != c.end(); col++)
				{
					int coltyp;
					std::string colna,colval("");
					colna = col->name();
					if(!col->is_null())
					{
						colval = col->c_str();
					}
					m_row.insert(std::pair<std::string, std::string>(topdbs_string::to_lower(colna), colval));
				}
				m_sqlrows++;
				break;
			}
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();		
			dbslog(DBS_ERROR,"ExecuteQuery_Row Exception  erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}	
		return m_sqlrows == 0 ? ERR_DBS_NOTFOUND : ERR_DBS_OK;	
	}

	int pqxx_session::ExecuteQuery(const std::string& sSQL)
	{
		stringmap_row_t row_data;
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		CHK_SQL_ERROR_RETURN(sSQL,ERR_DBS_ARGS);
		if(!IsSelectSQL(sSQL))
		{
			m_sqlerr = "Invalid SQL statement for Select";
			dbslog(DBS_ERROR,"ExecuteQuery Exception  erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}
		
		try
		{
			pqxx::connection* connection_ptr = (pqxx::connection*)m_pConnection->getConnection();
			pqxx::nontransaction N(*connection_ptr);
			pqxx::result R( N.exec( sSQL ));	

			m_sqlrows = R.size();
			m_res.clear();

			for(pqxx::result::const_iterator c = R.begin(); c != R.end(); ++c)
			{
				row_data.clear();
				for(auto col = c.begin(); col != c.end(); col++)
				{
					int coltyp;
					std::string colna,colval("");
					colna = col->name();
					if(!col->is_null())
					{
						colval = col->c_str();
					}
					row_data.insert(std::pair<std::string, std::string>(topdbs_string::to_lower(colna), colval));
				}
				m_res.push_back(row_data);
			}		
			m_row.clear();
			m_data_index = 0;					
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();		
			dbslog(DBS_ERROR,"ExecuteQuery Exception erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;
		}
		
		m_isResultSet = true;
		return ERR_DBS_OK;
	}
}
